package main

import (
	"coolpy7_benchmark/src/client"
	"coolpy7_benchmark/src/packet"
	"flag"
	"fmt"
	"log"
	"os"
	"os/signal"
	"strconv"
	"syscall"
	"time"
)

var urlString = flag.String("url", "tcp://username:password@127.0.0.1:1883", "broker url")
var topic = flag.String("topic", "$share/group1", "pub topic")
var workers = flag.Int("workers", 300, "number of workers")
//订阅者和链接数量一致
var receivers = flag.Int("receivers", 300, "number of receivers, same to workers")
var publishers = flag.Int("publishers", 100, "number of publishers")
var interval = flag.Int("i", 10, "interval of connecting to the broker(ms)")
var interval_of_msg = flag.Int("I", 1000, "interval of publishing message(ms)")
var size = flag.Int("s", 4096, "payload size(byte)")
var cs = flag.String("cid", "client", "client id start with")
var qos = flag.Uint("qos", 0, "pub qos level")
var clearsession = flag.Bool("clear", true, "clear session")
var pingtime = flag.String("keepalive", "300s", "keepalive")

func main() {
	flag.Parse()

	clients := make(map[string]*client.Client)

	for i := 0; i < *workers; i++ {
		id := strconv.Itoa(i)
		if i%1000 == 0 {
			log.Println(id)
		}

		cl := client.New()
		cl.Callback = func(msg *packet.Message, err error) error {
			if err != nil {
				log.Println("callback", err)
			}
			// if id != "0" {
			// 	log.Println(id)
			// }

			return nil
		}
		cf, err := cl.Connect(&client.Config{
			BrokerURL:    *urlString,
			CleanSession: *clearsession,
			KeepAlive:    *pingtime,
			ValidateSubs: true,
			ClientID:     *cs + id,
		})
		if err != nil {
			log.Println("conn", err)
		}
		//设置超时时间，不是中断等待，表示2秒如果conn没有返回就失败
		err = cf.Wait(2 * time.Second)
		if err != nil {
			log.Println("conn wait", err)
		}

		if *topic != "null" {
			tp := *topic+id
			sf, err := cl.Subscribe(tp, uint8(*qos))
			if err != nil {
				log.Println("sub", err)
			}
			//设置超时时间，不是中断等待
			err = sf.Wait(2 * time.Second)
			if err != nil {
				log.Println("sub wait", err)
			}
		}

		clients[*cs+id] = cl
		time.Sleep(time.Duration(*interval) * time.Millisecond)
	}
	fmt.Println("finish workers : ", *workers)
	fmt.Println("finish receivers : ", *receivers)
	
	//*size长度的字节，后面的参数是预留空间，并不是增加或者乘积
	bigData := make([]byte, *size, *size)

//	for i := 0; i < *workers; i++ {
//		go func() {
//			for _, v := range clients {
//				_, _ = v.Publish(*topic, bigData, uint8(*qos), false)
//			}
//		}()
//	}

	for j:=0; j<*publishers; j++{
		id := strconv.Itoa(j)
		cl := clients[*cs+id]
		go func() {
			for true {
				tp := *topic+id
				cl.Publish(tp, bigData, uint8(*qos), false)
				time.Sleep(time.Duration(*interval_of_msg) * time.Millisecond)
			}
		}()	
		//错开时间 *interval_of_msg * 1000 / *publishers
		time.Sleep(150 * time.Millisecond)
	}

	// for _, v := range clients {
	// 	go func() {
	// 		for true {
	// 			v.Publish(*topic, bigData, uint8(*qos), false)
	// 			time.Sleep(150 * time.Millisecond)
	// 		}
	// 	}()	
	// }

	fmt.Println("publishers", *publishers)

	signalChan := make(chan os.Signal, 1)
	cleanupDone := make(chan bool)
	signal.Notify(signalChan, syscall.SIGINT, syscall.SIGTERM)
	go func() {
		for range signalChan {
			cleanupDone <- true
		}
	}()
	<-cleanupDone
}
